"""执行器服务的Celery任务模块
提供任务执行的异步处理功能
"""
import os
import json
import time
import traceback
from datetime import datetime

from services.shared.celery_app import celery_app
from services.shared.database import Task, TaskExecution, db
from services.shared.db_pool import get_db_session, close_db_session
from services.shared.logger import get_logger
from services.executor.task_runner import run_task
from services.shared.models.enums import TaskStatus
from services.shared.timezone_utils import now as timezone_now
from services.shared.log_utils import log_error, log_info, log_system_event, log_task_operation
from services.shared.websocket_service import get_websocket_service
from config.config import get_config

# 获取日志记录器
logger = get_logger('executor.tasks')

# 初始化WebSocket服务URL（用于celery worker）
def _init_websocket_service():
    """初始化WebSocket服务URL"""
    try:
        websocket_service = get_websocket_service()
        # 在Docker环境中使用容器名，否则使用localhost
        if os.environ.get('SERVICE_TYPE') == 'celery-workers':
            config = get_config()
            web_service_url = f"http://xbyjob-app-services:{config.WEB_SERVICE_PORT}"
            websocket_service.set_web_service_url(web_service_url)
            logger.info(f"WebSocket服务URL设置为: {web_service_url}")
    except Exception as e:
        logger.warning(f"初始化WebSocket服务失败: {str(e)}")

# 在模块加载时初始化WebSocket服务
_init_websocket_service()


def _process_task_data(task_id=None, execution_id=None, task_data=None, **kwargs):
    """处理任务数据，确保格式一致
    
    Args:
        task_id: 任务ID
        execution_id: 执行ID
        task_data: 任务数据字典
        **kwargs: 其他参数
        
    Returns:
        处理后的任务数据字典
    """
    # 如果没有提供task_data或task_data不是字典，创建一个新字典
    if not task_data or not isinstance(task_data, dict):
        task_data = {}
    
    # 确保task_id和execution_id在task_data中
    if task_id is not None and 'task_id' not in task_data:
        task_data['task_id'] = task_id
    if execution_id is not None and 'execution_id' not in task_data:
        task_data['execution_id'] = execution_id
    
    # 将kwargs中的其他参数添加到task_data中
    for key, value in kwargs.items():
        if key not in ['task_id', 'execution_id', 'task_data'] and key not in task_data:
            task_data[key] = value
    
    return task_data


def _execute_task_once(task_id=None, execution_id=None, task_data=None, celery_task_id=None, retry_count=0, **kwargs):
    """执行单次任务的内部函数
    
    Args:
        task_id: 任务ID
        execution_id: 执行ID
        task_data: 任务数据字典
        celery_task_id: Celery任务ID
        retry_count: 当前重试次数
        **kwargs: 其他参数
        
    Returns:
        执行结果字典
    """
    logger.info(f"开始执行任务 (重试次数: {retry_count}): 任务ID={task_id}, 执行ID={execution_id}")
    
    # 处理任务数据
    task_data = _process_task_data(task_id, execution_id, task_data, **kwargs)
    logger.debug(f"处理后的task_data: {task_data}")
    
    if not task_data:
        logger.error("未提供有效的任务数据")
        return {'success': False, 'error': '未提供有效的任务数据'}
    
    task_id = task_data.get('task_id')
    execution_id = task_data.get('execution_id')
    params = task_data.get('params')
    
    logger.info(f"开始执行任务: {task_id}, 执行ID: {execution_id}, 参数: {params}")
    logger.debug(f"任务完整数据: {json.dumps(task_data, ensure_ascii=False)}")
    
    # 初始化结果
    result = {
        'success': False,
        'task_id': task_id,
        'execution_id': execution_id,
        'celery_task_id': celery_task_id,
        'start_time': timezone_now().isoformat(),
        'end_time': None,
        'output': '',
        'error': '',
        'status': TaskStatus.RUNNING,
        'return_code': None,
        'retry_count': retry_count
    }
    
    try:
        # 记录完整的任务数据
        logger.debug(f"开始处理任务数据: {json.dumps(task_data, ensure_ascii=False)}")
        
        # 获取任务信息
        logger.debug(f"准备从数据库获取任务信息，任务ID: {task_id}")
        session = get_db_session()
        logger.debug(f"数据库会话已创建: {session}")
        
        # 检查会话状态，处理可能的PendingRollbackError
        try:
            # 测试会话是否可用
            session.execute('SELECT 1')
        except Exception as e:
            logger.warning(f"数据库会话状态异常，尝试回滚: {str(e)}")
            try:
                session.rollback()
            except Exception as rollback_error:
                logger.error(f"会话回滚失败: {str(rollback_error)}")
                # 关闭当前会话并获取新会话
                session.close()
                session = get_db_session()
                logger.info("已重新获取数据库会话")
        
        task = session.query(Task).get(task_id)
        
        if not task:
            result['error'] = f"任务不存在: {task_id}"
            result['status'] = TaskStatus.FAILED
            logger.error(result['error'])
            logger.debug(f"数据库中未找到任务ID: {task_id}")
            return result
            
        # 记录从数据库获取的任务信息
        logger.debug(f"从数据库获取的任务信息: {task.__dict__}")
        logger.debug(f"任务类型: {task.task_type}, 脚本路径: {task.script_file_path}, 工作目录: {task.working_directory}")
        
        # 为了兼容 task_runner.py 中的代码，添加 type 属性并进行类型映射
        # 将 TaskType 枚举映射到 task_runner.py 中期望的值
        type_mapping = {
            'python_script': 'PYTHON'
        }
        task_type_value = task.task_type.value if hasattr(task.task_type, 'value') else str(task.task_type)
        task.type = type_mapping.get(task_type_value, task_type_value)
        
        # 如果没有提供执行ID，使用celery_task_id作为执行ID
        if not execution_id:
            logger.debug(f"未提供执行ID，将使用celery_task_id: {celery_task_id}")
            execution_id = celery_task_id
            result['execution_id'] = execution_id
        
        # 查找现有执行记录
        logger.debug(f"使用执行ID: {execution_id}，查找现有执行记录")
        execution = session.query(TaskExecution).filter_by(execution_id=execution_id).first()
        
        if execution:
            logger.debug(f"找到现有执行记录: {execution.__dict__}")
            execution.status = TaskStatus.RUNNING
            execution.executor_pid = os.getpid()
            execution.celery_task_id = celery_task_id  # 保存Celery任务ID
            execution.retry_count = retry_count  # 更新重试次数
            # 确保started_at不为空，如果为空则设置为当前时间
            if not execution.started_at:
                execution.started_at = timezone_now()
                logger.debug(f"设置执行开始时间: {execution.started_at}")
            else:
                logger.debug(f"保留现有开始时间: {execution.started_at}")
            
            # 更新任务的最后开始执行时间
            task.last_started_at = timezone_now()
            session.commit()
            session.refresh(task)  # 刷新task对象以避免DetachedInstanceError
            logger.debug(f"执行记录已更新为运行状态，Celery任务ID: {celery_task_id}")
            logger.debug(f"更新任务最后开始执行时间: {task.last_started_at}")
            
            # 发送WebSocket状态更新
            try:
                websocket_service = get_websocket_service()
                websocket_service.emit_execution_status_change(
                    execution_id=execution_id,
                    status=TaskStatus.RUNNING.value,
                    started_at=execution.started_at.isoformat() if execution.started_at else None
                )
                logger.debug(f"已发送WebSocket状态更新: 执行ID={execution_id}, 状态=RUNNING")
            except Exception as ws_e:
                logger.warning(f"发送WebSocket状态更新失败: {str(ws_e)}")
        else:
            # 创建新的执行记录
            logger.debug(f"未找到执行ID为 {execution_id} 的执行记录，创建新记录")
            execution = TaskExecution(
                task_id=task_id,
                status=TaskStatus.RUNNING,
                executor_pid=os.getpid(),
                execution_id=execution_id,
                celery_task_id=celery_task_id,
                retry_count=retry_count,
                started_at=timezone_now()
                # 移除 params 参数，因为 TaskExecution 模型中没有此字段
            )
            logger.debug(f"创建新的执行记录: {execution.__dict__}")
            session.add(execution)
            session.commit()
            logger.debug(f"新执行记录已保存到数据库，执行ID: {execution.execution_id}，Celery任务ID: {celery_task_id}")
            
            # 发送WebSocket状态更新
            try:
                websocket_service = get_websocket_service()
                websocket_service.emit_execution_status_change(
                    execution_id=execution_id,
                    status=TaskStatus.RUNNING.value,
                    started_at=execution.started_at.isoformat() if execution.started_at else None
                )
                logger.debug(f"已发送WebSocket状态更新: 执行ID={execution_id}, 状态=RUNNING")
            except Exception as ws_e:
                logger.warning(f"发送WebSocket状态更新失败: {str(ws_e)}")
        
        # 重新获取task对象以避免DetachedInstanceError
        task = session.get(Task, task_id)
        if not task:
            result['error'] = f"任务不存在: {task_id}"
            result['status'] = TaskStatus.FAILED
            logger.error(result['error'])
            return result
            
        # 获取task的属性值
        task_script_file_path = task.script_file_path
        task_id_value = task.id
        task_task_type = task.task_type
        task_working_directory = task.working_directory
        task_environment_variables = task.environment_variables
        task_timeout = task.timeout
        task_name = getattr(task, 'name', 'unknown')
        
        # 更新任务的最后开始执行时间
        task.last_started_at = timezone_now()
        session.commit()
        logger.debug(f"更新任务最后开始执行时间: {task.last_started_at}")
        
        # 刷新task对象以避免DetachedInstanceError
        session.refresh(task)
        
        # 检查是否有自定义脚本路径
        custom_script_path = task_data.get('script_file_path')
        if custom_script_path:
            logger.info(f"使用自定义脚本路径: {custom_script_path}")
            logger.debug(f"自定义脚本路径详情: {custom_script_path}")
            # 临时替换任务的脚本路径
            original_script_path = task_script_file_path
            # 重新获取task对象并更新脚本路径
            task = session.get(Task, task_id_value)
            task.script_file_path = custom_script_path
            session.commit()
            session.refresh(task)
            logger.debug(f"已临时替换任务脚本路径，原路径: {original_script_path}")
        else:
            logger.info("使用任务原始脚本路径")
            logger.debug(f"任务原始脚本路径: {task_script_file_path}")
            original_script_path = None
            
        # 执行任务
        try:
            logger.info(f"调用run_task执行任务，任务ID: {task_id_value}, 任务类型: {task_task_type}, 参数: {params}")
            logger.debug(f"执行任务详细信息 - 工作目录: {task_working_directory}, 环境变量: {task_environment_variables}, 超时: {task_timeout}")
            # 设置 task.type 属性为 task.task_type，以便 run_task 函数使用
            task.type = task_task_type
            # 设置 execution_id 属性，以便 run_task 函数进行实时输出推送
            task.execution_id = execution_id
            task_result = run_task(task, params)
            logger.info(f"任务执行结果: {task_result}")
            logger.debug(f"任务执行详细结果 - 返回码: {task_result.get('return_code')}, 输出长度: {len(task_result.get('output', ''))}, 错误长度: {len(task_result.get('error', ''))}")
        except Exception as e:
            error_msg = f"执行任务时发生异常: {str(e)}"
            logger.error(error_msg)
            logger.debug(f"异常类型: {type(e).__name__}")
            logger.debug(f"异常详情: {traceback.format_exc()}")
            
            # 记录任务执行失败的操作日志
            log_task_operation(
                action='execute',
                task_id=task_id,
                task_name=task_name,
                user_id=None,
                execution_time=None,
                status_code=500,
                error_message=str(e),
                details=f"Celery任务执行失败: {str(e)}"
            )
            
            task_result = {
                'output': '',
                'error': error_msg,
                'return_code': -1
            }
        
        # 如果使用了自定义脚本路径，恢复原始脚本路径
        if original_script_path is not None:
            # 重新查询task对象以避免DetachedInstanceError
            task = session.query(Task).get(task_id)
            if task:
                task.script_file_path = original_script_path
                session.commit()
        
        # 检查是否超时
        is_timeout = task_result.get('timeout', False)
        
        # 更新结果
        result.update({
            'success': task_result.get('success', False),
            'output': task_result.get('output', ''),
            'error': task_result.get('error', ''),
            'return_code': task_result.get('return_code'),
            'end_time': timezone_now().isoformat(),
            'status': TaskStatus.SUCCESS.value if task_result.get('success', False) else TaskStatus.FAILED.value,
            'timeout': is_timeout
        })
        
        # 更新执行记录
        try:
            logger.debug(f"准备更新执行记录，执行ID: {execution_id}")
            execution = session.query(TaskExecution).filter_by(execution_id=execution_id).first()
            if execution:
                logger.debug(f"找到执行记录: {execution.__dict__}")
                # 将字符串状态转换为TaskStatus枚举
                if result['status'] == TaskStatus.SUCCESS.value:
                    execution.status = TaskStatus.SUCCESS
                elif result['status'] == TaskStatus.FAILED.value:
                    execution.status = TaskStatus.FAILED
                else:
                    # 默认情况，保持原状态或设为FAILED
                    execution.status = TaskStatus.FAILED
                    logger.warning(f"未知的任务状态: {result['status']}，设置为FAILED")
                # 确保输出内容正确编码并限制长度
                if result['output']:
                    try:
                        # 确保输出是有效的UTF-8字符串
                        output = result['output']
                        if isinstance(output, bytes):
                            output = output.decode('utf-8', errors='replace')
                        # 确保输出内容是UTF-8编码
                        if isinstance(output, str):
                            # 直接使用字符串，不进行额外的编码转换
                            pass
                        execution.stdout = output[:65535] if len(output) > 65535 else output
                        logger.debug(f"处理后的stdout内容类型: {type(execution.stdout)}")
                        logger.debug(f"处理后的stdout内容: {execution.stdout}")
                    except Exception as e:
                        logger.error(f"处理stdout时出错: {str(e)}")
                        logger.debug(f"异常详情: {traceback.format_exc()}")
                        execution.stdout = "[输出内容编码错误]"
                
                if result['error']:
                    try:
                        # 确保错误是有效的UTF-8字符串
                        error = result['error']
                        if isinstance(error, bytes):
                            error = error.decode('utf-8', errors='replace')
                        # 确保错误输出内容是UTF-8编码
                        if isinstance(error, str):
                            # 直接使用字符串，不进行额外的编码转换
                            pass
                        
                        # 如果是超时，在错误信息前添加超时标识
                        if result.get('timeout', False):
                            error = f"[TIMEOUT] {error}"
                            
                        execution.stderr = error[:65535] if len(error) > 65535 else error
                        execution.error_message = error[:65535] if len(error) > 65535 else error
                        logger.debug(f"处理后的stderr内容类型: {type(execution.stderr)}")
                        logger.debug(f"处理后的stderr内容: {execution.stderr}")
                    except Exception as e:
                        logger.error(f"处理stderr时出错: {str(e)}")
                        logger.debug(f"异常详情: {traceback.format_exc()}")
                        execution.stderr = "[错误内容编码错误]"
                        execution.error_message = "[错误内容编码错误]"
                execution.exit_code = result['return_code']  # 使用exit_code字段存储返回码
                execution.completed_at = timezone_now()
                # 确保started_at不为空
                if not execution.started_at:
                    execution.started_at = timezone_now()
                    logger.warning(f"执行记录 {execution.execution_id} 的started_at为空，已设置为当前时间")
                
                # 计算任务执行时长
                # 确保两个datetime对象的时区一致
                started_at = execution.started_at
                completed_at = execution.completed_at
                
                # 统一转换为naive datetime进行计算，避免时区混合问题
                if started_at.tzinfo is not None:
                    started_at = started_at.replace(tzinfo=None)
                
                if completed_at.tzinfo is not None:
                    completed_at = completed_at.replace(tzinfo=None)
                
                duration = (completed_at - started_at).total_seconds()
                # 确保 duration 不为负数
                execution.duration = max(0, duration)
                
                session.commit()
                logger.debug(f"执行记录已更新，状态: {execution.status}, 时长: {execution.duration}秒")
                
                # 发送WebSocket状态和输出更新
                try:
                    websocket_service = get_websocket_service()
                    # 发送状态更新
                    websocket_service.emit_execution_status_change(
                        execution_id=execution_id,
                        status=execution.status.value,
                        completed_at=execution.completed_at.isoformat() if execution.completed_at else None,
                        duration=execution.duration,
                        exit_code=execution.exit_code
                    )
                    # 发送输出更新
                    if execution.stdout:
                        websocket_service.emit_execution_output(
                            execution_id=execution_id,
                            output_type='stdout',
                            content=execution.stdout
                        )
                    if execution.stderr:
                        websocket_service.emit_execution_output(
                            execution_id=execution_id,
                            output_type='stderr',
                            content=execution.stderr
                        )
                    logger.debug(f"已发送WebSocket完成状态和输出更新: 执行ID={execution_id}, 状态={execution.status.value}")
                except Exception as ws_e:
                    logger.warning(f"发送WebSocket完成更新失败: {str(ws_e)}")
            else:
                logger.warning(f"未找到执行ID为 {execution_id} 的执行记录，无法更新")
        except Exception as e:
            logger.error(f"更新执行记录失败: {str(e)}")
            logger.debug(f"异常详情: {traceback.format_exc()}")
            # 记录系统错误日志
            log_system_event(
                level='ERROR',
                module='executor',
                message=f"更新执行记录失败: {str(e)}",
                details=json.dumps({
                    "error": str(e),
                    "execution_id": execution_id,
                    "task_id": task_id,
                    "execution_context": {
                        "function": "_execute_task_once",
                        "module": "executor.tasks",
                        "operation": "update_execution_record",
                        "database_operation": "update_and_commit"
                    }
                }, ensure_ascii=False),
                action='UPDATE_EXECUTION_RECORD',
                resource_type='TASK_EXECUTION',
                resource_id=str(execution_id)
            )
        
        # 更新任务状态
        try:
            logger.debug(f"准备更新任务状态，任务ID: {task_id}")
            # 重新查询task对象以避免DetachedInstanceError
            task = session.query(Task).get(task_id)
            if not task:
                logger.error(f"无法找到任务ID: {task_id}")
                return result
                
            if result['success']:
                task.status = TaskStatus.SUCCESS
                task.last_run_at = timezone_now()
                logger.debug(f"任务执行成功，更新状态为SUCCESS")
            else:
                task.status = TaskStatus.FAILED
                logger.debug(f"任务执行失败，更新状态为FAILED")
                
                # 如果任务执行失败且启用了失败通知，发送钉钉通知
                if task.notify_on_failure and task.notification_config_id:
                    try:
                        from services.shared.notification_service import NotificationService
                        logger.debug(f"任务失败，准备发送通知：任务ID={task_id}, 配置ID={task.notification_config_id}")
                        
                        # 获取执行记录
                        execution_record = session.query(TaskExecution).filter_by(execution_id=execution_id).first()
                        if execution_record:
                            success, message = NotificationService.send_task_failure_notification_by_config_id(
                                task, execution_record, task.notification_config_id
                            )
                            if success:
                                logger.info(f"任务失败通知发送成功：任务ID={task_id}")
                            else:
                                logger.warning(f"任务失败通知发送失败：任务ID={task_id}, 错误={message}")
                        else:
                            logger.warning(f"未找到执行记录，无法发送失败通知：执行ID={execution_id}")
                    except Exception as notify_e:
                        logger.error(f"发送失败通知时出错：{str(notify_e)}")
                        logger.debug(f"通知异常详情: {traceback.format_exc()}")
            
            session.commit()
            logger.debug(f"任务状态已更新: {task.status}")
            logger.info(f"任务执行完成: {task_id}, 执行ID: {execution_id}, 状态: {result['status']}")
        except Exception as e:
            logger.error(f"更新任务状态失败: {str(e)}")
            logger.debug(f"异常详情: {traceback.format_exc()}")
            # 记录系统错误日志
            log_system_event(
                level='ERROR',
                module='executor',
                message=f"更新任务状态失败: {str(e)}",
                details=json.dumps({
                    "error": str(e),
                    "task_id": task_id,
                    "execution_context": {
                        "function": "_execute_task_once",
                        "module": "executor.tasks",
                        "operation": "update_task_status",
                        "database_operation": "update_and_commit"
                    }
                }, ensure_ascii=False),
                action='UPDATE_TASK_STATUS',
                resource_type='TASK',
                resource_id=str(task_id)
            )
        finally:
            logger.debug(f"关闭数据库会话")
            try:
                if session:
                    session.close()
            except Exception as e:
                logger.error(f"关闭数据库会话失败: {str(e)}")
            # 不需要调用close_db_session()，因为session.close()已经足够
    
    except Exception as e:
        error_msg = f"任务执行异常: {str(e)}\n{traceback.format_exc()}"
        logger.error(error_msg)
        
        # 记录详细的系统错误日志
        log_task_operation(
            action='execute',
            task_id=task_id,
            task_name=f"Task-{task_id}",
            user_id=None,
            execution_time=None,
            status_code=500,
            error_message=f"Celery任务严重失败: {str(e)}",
            details=f"任务执行异常: {error_msg}"
        )
        
        # 更新结果
        result.update({
            'success': False,
            'error': error_msg,
            'end_time': datetime.utcnow().isoformat(),
            'status': TaskStatus.FAILED.value
        })
        
        # 更新执行记录和任务状态
        try:
            session = get_db_session()
            execution = session.query(TaskExecution).filter_by(execution_id=execution_id).first()
            if execution:
                execution.status = TaskStatus.FAILED
                execution.error_message = error_msg
                execution.retry_count = retry_count
                execution.completed_at = timezone_now()
                # 确保started_at不为空
                if not execution.started_at:
                    execution.started_at = timezone_now()
                    logger.warning(f"异常情况下执行记录 {execution.execution_id} 的started_at为空，已设置为当前时间")
                
                # 计算任务执行时长
                duration = (execution.completed_at - execution.started_at).total_seconds()
                # 确保 duration 不为负数
                execution.duration = max(0, duration)
                
                session.commit()
                logger.debug(f"异常情况下执行记录已更新")
                
                # 发送WebSocket状态更新
                try:
                    websocket_service = get_websocket_service()
                    websocket_service.emit_execution_status_change(
                        execution_id=execution_id,
                        status=TaskStatus.FAILED.value,
                        completed_at=execution.completed_at.isoformat() if execution.completed_at else None,
                        duration=execution.duration,
                        error_message=execution.error_message
                    )
                    logger.debug(f"已发送WebSocket异常状态更新: 执行ID={execution_id}, 状态=FAILED")
                except Exception as ws_e:
                    logger.warning(f"发送WebSocket异常状态更新失败: {str(ws_e)}")
            
            # 更新任务状态
            task = session.query(Task).get(task_id)
            if task:
                task.status = TaskStatus.FAILED
                session.commit()
                logger.debug(f"异常情况下任务状态已更新为FAILED")
        except Exception as update_error:
            logger.error(f"异常情况下更新数据库失败: {str(update_error)}")
            logger.debug(f"异常详情: {traceback.format_exc()}")
            # 记录系统错误日志
            log_system_event(
                level='ERROR',
                module='executor',
                message=f"异常情况下更新数据库失败: {str(update_error)}",
                details=json.dumps({
                    "error": str(update_error),
                    "original_error": str(e),
                    "execution_id": execution_id,
                    "task_id": task_id,
                    "execution_context": {
                        "function": "_execute_task_once",
                        "module": "executor.tasks",
                        "operation": "update_failed_task_execution_record",
                        "database_operation": "update_and_commit"
                    }
                }, ensure_ascii=False),
                action='UPDATE_EXECUTION_RECORD',
                resource_type='TASK_EXECUTION',
                resource_id=str(execution_id)
            )
        finally:
            try:
                if session:
                    session.close()
            except Exception as e:
                logger.error(f"关闭数据库会话失败: {str(e)}")
    
    return result


@celery_app.task(bind=True, name='services.executor.tasks.execute_task')
def execute_task(self, task_id=None, execution_id=None, task_data=None, **kwargs):
    """执行任务（带自动重试功能）
    
    支持多种调用方式：
    1. execute_task(task_data=任务数据字典)
    2. execute_task(task_id=任务ID, execution_id=执行ID, **其他参数)
    3. execute_task(task_id=任务ID, execution_id=执行ID, task_data=任务数据字典)
    
    Args:
        self: Celery任务实例
        task_data: 任务数据字典
        task_id: 任务ID
        execution_id: 执行ID
        **kwargs: 其他参数
    
    Returns:
        执行结果字典
    """
    # 增加更详细的日志记录
    logger.info(f"任务执行开始: 任务ID={task_id}, 执行ID={execution_id}, 数据={task_data}, 其他参数={kwargs}")
    logger.info(f"Celery任务ID: {self.request.id if hasattr(self, 'request') else 'unknown'}")
    
    # 记录接收到的参数，用于调试
    logger.debug(f"execute_task接收到的参数: task_data={task_data}, task_id={task_id}, execution_id={execution_id}, kwargs={kwargs}")
    logger.debug(f"Celery任务实例ID: {self.request.id if hasattr(self, 'request') else 'unknown'}")
    
    # 获取Celery任务ID
    try:
        # 尝试获取Celery任务ID
        from celery import current_task
        celery_task_id = current_task.request.id if current_task else 'unknown'
        logger.debug(f"获取到Celery任务ID: {celery_task_id}")
    except Exception as e:
        logger.warning(f"无法获取Celery任务ID: {str(e)}")
        logger.debug(f"获取Celery任务ID异常详情: {traceback.format_exc()}")
        celery_task_id = 'unknown'
    
    # 处理任务数据
    task_data = _process_task_data(task_id, execution_id, task_data, **kwargs)
    logger.debug(f"处理后的task_data: {task_data}")
    
    if not task_data:
        logger.error("未提供有效的任务数据")
        return {'success': False, 'error': '未提供有效的任务数据'}
    
    task_id = task_data.get('task_id')
    execution_id = task_data.get('execution_id')
    
    # 获取配置
    config = get_config()
    default_retry_attempts = getattr(config, 'RETRY_ATTEMPTS', 3)
    retry_delay = getattr(config, 'RETRY_DELAY', 60)
    
    # 获取任务的重试配置
    max_retry_attempts = default_retry_attempts  # 默认使用全局配置
    try:
        session = get_db_session()
        task = session.query(Task).get(task_id)
        if task:
            # 如果任务配置了重试次数，则使用任务配置（包括0次）
            # 如果任务的retry_attempts为0，表示不重试
            max_retry_attempts = task.retry_attempts
        session.close()
    except Exception as e:
        logger.warning(f"获取任务重试配置失败，使用默认配置: {str(e)}")
    
    logger.info(f"任务重试配置: 最大重试次数={max_retry_attempts}, 重试延迟={retry_delay}秒")
    
    # 执行任务（带重试逻辑）
    retry_count = 0
    last_result = None
    
    while retry_count <= max_retry_attempts:
        logger.info(f"开始第 {retry_count + 1} 次执行任务 (最大重试次数: {max_retry_attempts})")
        
        # 执行单次任务
        result = _execute_task_once(
            task_id=task_id,
            execution_id=execution_id,
            task_data=task_data,
            celery_task_id=celery_task_id,
            retry_count=retry_count,
            **kwargs
        )
        
        last_result = result
        
        # 如果任务执行成功，直接返回结果
        if result.get('success', False):
            logger.info(f"任务执行成功 (第 {retry_count + 1} 次尝试)")
            return result
        
        # 如果任务失败且还有重试机会
        if retry_count < max_retry_attempts:
            retry_count += 1
            logger.warning(f"任务执行失败，将在 {retry_delay} 秒后进行第 {retry_count + 1} 次重试")
            logger.warning(f"失败原因: {result.get('error', '未知错误')}")
            
            # 等待重试延迟时间
            time.sleep(retry_delay)
        else:
            # 已达到最大重试次数
            logger.error(f"任务执行失败，已达到最大重试次数 ({max_retry_attempts + 1} 次)")
            break
    
    # 所有重试都失败了，返回最后一次的结果
    if last_result:
        last_result['final_retry_count'] = retry_count
        last_result['max_retry_attempts'] = max_retry_attempts
        logger.error(f"任务最终执行失败，总共尝试了 {retry_count + 1} 次")
    
    return last_result or {'success': False, 'error': '任务执行失败，无有效结果'}


@celery_app.task(name='cleanup_executions')
def cleanup_executions(days=30):
    """清理过期的执行记录
    
    Args:
        days: 保留天数，默认30天
    
    Returns:
        清理结果字典
    """
    from datetime import datetime, timedelta
    import time
    
    logger.info(f"开始清理{days}天前的执行记录")
    logger.debug(f"清理参数 - 保留天数: {days}")
    cutoff_date = timezone_now() - timedelta(days=days)
    logger.debug(f"清理截止日期: {cutoff_date}")
    
    try:
        logger.debug("准备获取数据库会话")
        session = get_db_session()
        # 分批删除以避免锁定数据库太长时间
        batch_size = 1000
        logger.debug(f"设置批处理大小: {batch_size}")
        total_deleted = 0
        batch_count = 0
        
        while True:
            logger.debug(f"开始处理第{batch_count + 1}批数据")
            executions = session.query(TaskExecution).filter(TaskExecution.started_at < cutoff_date).limit(batch_size).all()
            logger.debug(f"查询到{len(executions)}条需要删除的执行记录")
            
            if not executions:
                logger.debug("没有更多需要删除的执行记录，清理完成")
                break
                
            logger.debug(f"开始删除第{batch_count + 1}批数据，共{len(executions)}条记录")
            for execution in executions:
                logger.debug(f"删除执行记录 - ID: {execution.id}, 执行ID: {execution.execution_id}, 创建时间: {execution.created_at}")
                session.delete(execution)
            
            logger.debug("提交删除操作到数据库")
            session.commit()
            total_deleted += len(executions)
            batch_count += 1
            
            logger.debug(f"已完成第{batch_count}批数据删除，当前共删除{total_deleted}条记录")
            # 短暂暂停，避免数据库负载过高
            time.sleep(0.1)
            
            logger.info(f"清理完成，共删除{total_deleted}条执行记录")
            logger.debug(f"清理统计 - 总批次: {batch_count}, 总删除记录: {total_deleted}")
            session.close()
            close_db_session()
            return {'success': True, 'deleted_count': total_deleted, 'batch_count': batch_count}
            
    except Exception as e:
        error_msg = f"清理执行记录失败: {str(e)}"
        logger.error(error_msg)
        logger.debug(f"清理异常详情: {traceback.format_exc()}")
        
        # 记录清理执行记录失败的系统事件日志
        log_system_event(
            level='ERROR',
            module='executor',
            message=f"清理执行记录失败: {str(e)}",
            details=json.dumps({
                "days": days,
                "error_type": type(e).__name__,
                "error_message": str(e),
                "stack_trace": traceback.format_exc(),
                "execution_context": {
                    "function": "cleanup_executions",
                    "module": "executor.tasks",
                    "operation": "cleanup_old_task_executions",
                    "cleanup_type": "batch_delete"
                }
            }, ensure_ascii=False),
            action='CLEANUP_EXECUTIONS',
            resource_type='TASK_EXECUTION'
        )
        
        return {'success': False, 'error': error_msg}


@celery_app.task(bind=True, name='services.executor.tasks.stop_task')
def stop_task(self, execution_id):
    """停止正在运行的任务
    
    Args:
        self: Celery任务实例
        execution_id: 执行ID
        
    Returns:
        停止结果字典
    """
    logger.info(f"开始停止任务: 执行ID={execution_id}")
    
    try:
        # 获取数据库会话
        session = get_db_session()
        
        # 查找执行记录
        execution = session.query(TaskExecution).filter_by(execution_id=execution_id).first()
        if not execution:
            logger.warning(f"未找到执行记录: {execution_id}")
            return {'success': False, 'error': f'执行记录不存在: {execution_id}'}
        
        # 检查任务状态
        if execution.status not in [TaskStatus.PENDING, TaskStatus.RUNNING]:
            logger.info(f"任务状态为{execution.status.value}，无需停止")
            return {'success': True, 'message': f'任务状态为{execution.status.value}，无需停止'}
        
        # 如果有Celery任务ID，尝试撤销任务
        if hasattr(execution, 'celery_task_id') and execution.celery_task_id:
            logger.info(f"撤销Celery任务: {execution.celery_task_id}")
            try:
                celery_app.control.revoke(execution.celery_task_id, terminate=True)
                logger.info(f"Celery任务已撤销: {execution.celery_task_id}")
            except Exception as e:
                # CANCELLED异常是正常的，表示任务已被取消
                if 'CANCELLED' in str(e).upper():
                    logger.info(f"Celery任务已被取消: {execution.celery_task_id}")
                else:
                    logger.warning(f"撤销Celery任务失败: {str(e)}")
        
        # 如果有进程ID，尝试终止进程
        if hasattr(execution, 'executor_pid') and execution.executor_pid:
            logger.info(f"尝试终止进程: {execution.executor_pid}")
            try:
                import signal
                os.kill(execution.executor_pid, signal.SIGTERM)
                logger.info(f"进程已终止: {execution.executor_pid}")
            except ProcessLookupError:
                logger.info(f"进程不存在或已结束: {execution.executor_pid}")
            except Exception as e:
                logger.warning(f"终止进程失败: {str(e)}")
        
        # 更新执行记录状态
        execution.status = TaskStatus.CANCELLED
        execution.completed_at = timezone_now()
        execution.error_message = '任务被手动停止'
        
        # 提交数据库更改
        session.commit()
        logger.info(f"任务状态已更新为CANCELLED: {execution_id}")
        
        # 记录系统事件日志
        log_system_event(
            level='INFO',
            module='executor',
            message=f"任务执行已停止: {execution_id}",
            details=json.dumps({
                "execution_id": execution_id,
                "task_id": execution.task_id,
                "previous_status": execution.status.value if hasattr(execution, 'status') else 'unknown',
                "celery_task_id": getattr(execution, 'celery_task_id', None),
                "executor_pid": getattr(execution, 'executor_pid', None)
            }, ensure_ascii=False),
            action='STOP_TASK',
            resource_type='TASK_EXECUTION',
            resource_id=execution_id
        )
        
        logger.info(f"任务停止成功: {execution_id}")
        return {'success': True, 'message': '任务已停止'}
        
    except Exception as e:
        # 如果异常是CANCELLED，这通常表示任务已被成功取消
        if 'CANCELLED' in str(e).upper():
            logger.info(f"任务已被取消: {execution_id}")
            try:
                # 重新获取数据库会话和执行记录来更新状态
                new_session = get_db_session()
                execution_record = new_session.query(TaskExecution).filter_by(execution_id=execution_id).first()
                if execution_record:
                    execution_record.status = TaskStatus.CANCELLED
                    execution_record.completed_at = timezone_now()
                    execution_record.error_message = '任务被手动停止'
                    new_session.commit()
                    logger.info(f"任务状态已更新为CANCELLED: {execution_id}")
                else:
                    logger.warning(f"未找到执行记录: {execution_id}")
                new_session.close()
            except Exception as db_e:
                logger.warning(f"更新已取消任务状态失败: {str(db_e)}")
                logger.error(f"数据库更新异常详情: {traceback.format_exc()}")
                # 尝试回滚会话
                try:
                    if 'new_session' in locals():
                        new_session.rollback()
                        new_session.close()
                except Exception as rollback_e:
                    logger.error(f"回滚数据库会话失败: {str(rollback_e)}")
            return {'success': True, 'message': '任务已停止'}
        
        error_msg = f"停止任务时发生异常: {str(e)}"
        logger.error(error_msg)
        logger.debug(f"停止任务异常详情: {traceback.format_exc()}")
        
        # 记录错误日志
        log_error(
            "executor.tasks",
            f'停止任务失败: {str(e)}',
            details=json.dumps({
                "execution_id": execution_id,
                "function": "stop_task",
                "error_type": "STOP_TASK_ERROR",
                "stack_trace": traceback.format_exc()
            }, ensure_ascii=False)
        )
        
        return {'success': False, 'error': error_msg}
    
    finally:
        # 确保关闭数据库会话
        try:
            if 'session' in locals() and session:
                session.close()
        except Exception as e:
            logger.error(f"关闭数据库会话失败: {str(e)}")


@celery_app.task(bind=True, name='services.executor.tasks.execute_script_test')
def execute_script_test(self, test_execution_id, script_name, script_content, script_language, username, user_id):
    """执行脚本测试任务
    
    Args:
        self: Celery任务实例
        test_execution_id: 测试执行ID
        script_name: 脚本名称
        script_content: 脚本内容
        script_language: 脚本语言 (python, bash, nodejs, java)
        username: 用户名
        user_id: 用户ID
        
    Returns:
        执行结果字典
    """
    import tempfile
    import subprocess
    import uuid
    from pathlib import Path
    
    # 使用传入的测试执行ID
    test_id = test_execution_id
    
    logger.info(f"开始执行脚本测试: 脚本名称={script_name}, 测试ID={test_id}, 语言={script_language}, 用户={username}")
    
    try:
        # 获取WebSocket服务
        websocket_service = get_websocket_service()
        
        # 发送测试开始状态
        websocket_service.emit_script_test_output(
            test_id=test_id,
            output_type='status',
            content='测试开始执行...\n'
        )
        
        # 创建临时文件
        with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix=f'.{_get_file_extension(script_language)}') as temp_file:
            temp_file.write(script_content)
            temp_file_path = temp_file.name
        
        try:
            # 构建执行命令
            command = _build_test_command(script_language, temp_file_path)
            
            logger.info(f"执行命令: {' '.join(command)}")
            
            # 设置环境变量
            env = os.environ.copy()
            env['PYTHONUNBUFFERED'] = '1'
            env['PYTHONIOENCODING'] = 'utf-8'
            
            # 添加公共方法目录到PYTHONPATH（仅对Python脚本）
            if script_language == 'python':
                from config.config import get_config
                config = get_config()
                common_methods_dir = os.path.abspath(config.COMMON_METHODS_DIR)
                
                # 将公共方法目录添加到PYTHONPATH
                current_pythonpath = env.get('PYTHONPATH', '')
                if current_pythonpath:
                    env['PYTHONPATH'] = f"{common_methods_dir}:{current_pythonpath}"
                else:
                    env['PYTHONPATH'] = common_methods_dir
                
                logger.info(f"已将公共方法目录添加到PYTHONPATH: {common_methods_dir}")
                
                # 通过WebSocket发送PYTHONPATH设置信息
                websocket_service.emit_script_test_output(
                    test_id=test_id,
                    output_type='status',
                    content=f'已设置PYTHONPATH包含公共方法目录: {common_methods_dir}\n'
                )
            
            # 注入公共参数到环境变量
            if user_id:
                try:
                    from services.shared.models import GlobalParameter
                    from services.shared.db_pool import get_db_session
                    from services.shared.crypto_utils import decrypt_value
                    
                    # 获取数据库会话
                    db_session = get_db_session()
                    
                    # 获取用户的所有启用的公共参数
                    global_params = db_session.query(GlobalParameter).filter(
                        GlobalParameter.user_id == user_id,
                        GlobalParameter.is_enabled == True
                    ).all()
                    
                    injected_count = 0
                    # 将公共参数设置为环境变量，添加GLOBAL_前缀
                    for param in global_params:
                        env_key = f"GLOBAL_{param.name}"
                        env_value = param.value
                        
                        # 如果是加密参数，进行解密
                        if param.is_encrypted:
                            try:
                                env_value = decrypt_value(param.value)
                                logger.info(f"成功解密公共参数: {param.name}")
                            except Exception as decrypt_error:
                                logger.error(f"解密公共参数 {param.name} 失败: {str(decrypt_error)}")
                                # 解密失败时跳过该参数
                                continue
                        
                        env[env_key] = str(env_value)
                        injected_count += 1
                        # 对于加密参数，不在日志中显示实际值
                        if param.is_encrypted:
                            logger.info(f"设置加密公共参数环境变量: {env_key} = [已加密]")
                        else:
                            logger.info(f"设置公共参数环境变量: {env_key} = {env_value}")
                    
                    logger.info(f"成功注入 {injected_count} 个公共参数到测试执行环境")
                    
                    # 通过WebSocket发送公共参数注入信息
                    if injected_count > 0:
                        websocket_service.emit_script_test_output(
                            test_id=test_id,
                            output_type='status',
                            content=f'已注入 {injected_count} 个公共参数到执行环境\n'
                        )
                    
                except Exception as e:
                    logger.warning(f"注入公共参数失败: {str(e)}，将继续执行脚本测试")
                    websocket_service.emit_script_test_output(
                        test_id=test_id,
                        output_type='status',
                        content=f'注入公共参数失败: {str(e)}，将继续执行脚本测试\n'
                    )
            
            # 执行脚本
            process = subprocess.Popen(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,
                universal_newlines=True,
                env=env
            )
            
            # 实时读取输出
            import threading
            
            def read_output(pipe, output_type):
                """读取进程输出并通过WebSocket发送"""
                try:
                    for line in iter(pipe.readline, ''):
                        if line:
                            websocket_service.emit_script_test_output(
                                test_id=test_id,
                                output_type=output_type,
                                content=line
                            )
                    pipe.close()
                except Exception as e:
                    logger.error(f"读取{output_type}输出失败: {str(e)}")
            
            # 启动输出读取线程
            stdout_thread = threading.Thread(target=read_output, args=(process.stdout, 'stdout'))
            stderr_thread = threading.Thread(target=read_output, args=(process.stderr, 'stderr'))
            
            stdout_thread.start()
            stderr_thread.start()
            
            # 等待进程完成
            return_code = process.wait()
            
            # 等待输出线程完成
            stdout_thread.join(timeout=5)
            stderr_thread.join(timeout=5)
            
            # 发送完成状态
            if return_code == 0:
                websocket_service.emit_script_test_output(
                    test_id=test_id,
                    output_type='status',
                    content=f'\n测试执行完成，退出码: {return_code}\n'
                )
                logger.info(f"脚本测试执行成功: 脚本名称={script_name}, 测试ID={test_id}")
            else:
                websocket_service.emit_script_test_output(
                    test_id=test_id,
                    output_type='status',
                    content=f'\n测试执行失败，退出码: {return_code}\n'
                )
                logger.warning(f"脚本测试执行失败: 脚本名称={script_name}, 测试ID={test_id}, 退出码={return_code}")
            
            return {
                'success': return_code == 0,
                'return_code': return_code,
                'test_id': test_id,
                'script_name': script_name
            }
            
        finally:
            # 清理临时文件
            try:
                os.unlink(temp_file_path)
                logger.debug(f"已删除临时文件: {temp_file_path}")
            except Exception as e:
                logger.warning(f"删除临时文件失败: {str(e)}")
                
    except Exception as e:
        error_msg = f"脚本测试执行异常: {str(e)}"
        logger.error(error_msg)
        logger.debug(f"异常详情: {traceback.format_exc()}")
        
        # 发送错误状态
        try:
            websocket_service = get_websocket_service()
            websocket_service.emit_script_test_output(
                test_id=test_id,
                output_type='error',
                content=f'\n执行异常: {error_msg}\n'
            )
        except Exception as ws_e:
            logger.error(f"发送WebSocket错误消息失败: {str(ws_e)}")
        
        return {
            'success': False,
            'error': error_msg,
            'test_id': test_id,
            'script_name': script_name
        }


def _get_file_extension(language):
    """根据语言获取文件扩展名"""
    extensions = {
        'python': 'py',
        'bash': 'sh',
        'nodejs': 'js',
        'java': 'java'
    }
    return extensions.get(language, 'txt')


def _build_test_command(language, file_path):
    """根据语言构建执行命令"""
    commands = {
        'python': ['python3', file_path],
        'bash': ['bash', file_path],
        'nodejs': ['node', file_path],
        'java': ['java', file_path]  # 简化处理，实际可能需要编译
    }
    
    command = commands.get(language)
    if not command:
        raise ValueError(f"不支持的脚本语言: {language}")
    
    return command